package com.niit.qcx.PriceType;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**数据中包含的字段
 * title comm  total_price unit_price  no_room  area  orientations  floor  year  address	county	town  tags  broker	rate  company
 * 研究不同价格区间房源的主要户型构成
 *Mapper阶段：
 *提取价格（总价total_price）和户型（no_room）
 *根据包含数值大致范围进行价格区间划分，例如 0 ~ 500、501 ~ 1000 等区间，定义方法将单价与区间上下限比较确定所在区间并返回
 * 将价格区间作为键，主要户型作为值输出。例如，输出<0 ~ 500 , 3室2厅2卫>。
 */

public class PriceTypeMapper extends Mapper<LongWritable, Text, Text, Text> {

    private static String[] priceRanges = {"0~500", "501~1000", "1001~1500","1501~2000","2001~2500","2501~3000","3001~3500","3501~4000","4000~15000"};

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();
        String[] fields = line.split("\t");

        if (key.get() == 0) {// 跳过第一行标题
            return;
        }

        if (fields.length < 5) {
            return; // 忽略格式不正确的行
        }

        if (fields[2].isEmpty() || fields[4].isEmpty()){
            return;//忽略缺少所需字段的行
        }

        float total_price = Float.parseFloat(fields[2]);

        String no_room = fields[4];
        String priceRange = getPriceRange(total_price);

        if (priceRange != null) {
            context.write(new Text(priceRange), new Text(no_room));
        }
    }

    private String getPriceRange(float total_price) {

        for (String range : priceRanges) {

            String[] bounds = range.split("~");

            int lowerBound = Integer.parseInt(bounds[0]);
            int upperBound = Integer.parseInt(bounds[1]);

            if (total_price >= lowerBound && total_price <= upperBound) {
                return range;
            }


        }
        return null;
    }
}
